package com.jxw.cloudpen.web.netty.client;

import com.google.api.gax.rpc.ClientStream;
import com.google.api.gax.rpc.ResponseObserver;
import com.google.api.gax.rpc.StreamController;
import com.google.cloud.speech.v1p1beta1.*;
import com.google.protobuf.ByteString;
import com.google.protobuf.Duration;
import com.jxw.cloudpen.web.netty.Entity.StreamData;
import com.jxw.cloudpen.web.netty.common.CommonRpcDecoderHandler;
import com.jxw.cloudpen.web.netty.common.CommonRpcEncoderHandler;
import com.jxw.cloudpen.web.netty.handler.BusinHandler;
import com.jxw.cloudpen.web.netty.handler.LoginResponceHandler;
import com.jxw.cloudpen.web.netty.util.ByteObjUtil;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.DefaultEventExecutorGroup;
import io.netty.util.concurrent.EventExecutorGroup;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.sound.sampled.*;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;


public class NettyClientBootstrap {
    Logger log = LoggerFactory.getLogger(NettyClientBootstrap.class);

    public static String clientId;
    private int port;
    private String host;
    private Channel channel;
    private static final EventExecutorGroup group = new DefaultEventExecutorGroup(20);

    public NettyClientBootstrap(int port, String host) throws InterruptedException {
        this.port = port;
        this.host = host;
        start();
    }

    public Channel getChannel() {
        return channel;
    }

    private void start() throws InterruptedException {
        EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.channel(NioSocketChannel.class);
//        bootstrap.option(ChannelOption.SO_KEEPALIVE, true);


        bootstrap.option(ChannelOption.TCP_NODELAY, true);

        bootstrap.group(eventLoopGroup);
        bootstrap.remoteAddress(host, port);
        bootstrap.handler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel socketChannel) throws Exception {
                socketChannel.pipeline().addLast(new IdleStateHandler(20, 10, 0));

                socketChannel.pipeline().addLast("decoder", new CommonRpcDecoderHandler());
                socketChannel.pipeline().addLast("encoder", new CommonRpcEncoderHandler());
                socketChannel.pipeline().addLast("loginResponseHandler",new LoginResponceHandler());
//                socketChannel.pipeline().addLast(new BusinHandler());
            }
        });
        ChannelFuture future = bootstrap.connect(host, port).sync();
        if (future.isSuccess()) {
            channel =future.channel();
            System.out.println("connect server  成功---------");
            //发送登录消息
            //登录请求
            StreamData streamData = new StreamData();
            streamData.setSecretKey("test");
            streamData.setType(0);
            channel.writeAndFlush(ByteObjUtil.objToBytes(streamData));
            Thread.sleep(1000L);
        }
    }

    public static void main(String[] args) throws InterruptedException {
        //Constants.setClientId("101");
        NettyClientBootstrap bootstrap = new NettyClientBootstrap(9999, "127.0.0.1");
        StreamData result = new StreamData();
        //执行录音
        try {
            executeMusic(bootstrap);
        } catch (LineUnavailableException e) {
            e.printStackTrace();
        }
//        bootstrap.channel.writeAndFlush();
    }
    private static TargetDataLine targetDataLine;

    private static volatile BlockingQueue<byte[]> sharedQueue = new LinkedBlockingQueue<byte[]>();


    public static StreamData executeMusic(NettyClientBootstrap bootstrap) throws LineUnavailableException {


        // SampleRate:16000Hz, SampleSizeInBits: 16, Number of channels: 1, Signed: true,
        // bigEndian: false
        AudioFormat audioFormat = new AudioFormat(16000, 16, 1, true, false);
        DataLine.Info targetInfo =
                new DataLine.Info(
                        TargetDataLine.class,
                        audioFormat); // Set the system information to read from the microphone audio
        // stream

        if (!AudioSystem.isLineSupported(targetInfo)) {
            System.out.println("Microphone not supported");
            System.exit(0);
        }
        // Target data line captures the audio stream the microphone produces.
        targetDataLine = (TargetDataLine) AudioSystem.getLine(targetInfo);
        targetDataLine.open(audioFormat);

        StreamData result = new StreamData();
// Microphone Input buffering
        class MicBuffer implements Runnable {

            @Override
            public void run() {
                System.out.println("Start speaking...Press Ctrl-C to stop");
                targetDataLine.start();
                byte[] data = new byte[6400];
                while (targetDataLine.isOpen()) {
                    try {
                        int numBytesRead = targetDataLine.read(data, 0, data.length);
                        if ((numBytesRead <= 0) && (targetDataLine.isOpen())) {
                            continue;
                        }
                        //组装数据
                        sharedQueue.put(data.clone());
                    } catch (InterruptedException e) {
                        System.out.println("Microphone input buffering interrupted : " + e.getMessage());
                    }
                }
            }
        }

        // Creating microphone input buffer thread
        MicBuffer micrunnable = new MicBuffer();
        Thread micThread = new Thread(micrunnable);
        micThread.start();
        ResponseObserver<StreamingRecognizeResponse> responseObserver = null;
        try {
            if (!AudioSystem.isLineSupported(targetInfo)) {
                System.out.println("Microphone not supported");
                System.exit(0);
            }
            long startTime = System.currentTimeMillis();
            while (true) {
                byte[] bytes=  sharedQueue.take();
                if(bytes!=null && bytes.length>0){
                    result.setClientId(clientId);
                    result.setData(bytes);
                    result.setType(6);
//                bootstrap.channel.writeAndFlush(result);
                    bootstrap.channel.writeAndFlush(ByteObjUtil.objToBytes(result));
                }

            }
        } catch (Exception e) {
            System.out.println(e);
        }

        return result;
    }
}
